package com.chis.jxc;

import com.alibaba.fastjson.JSON;
import com.chis.jm.bean.jxc.trace.TraceBean;
import com.chis.jxc.utils.DecodeingKafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

public class JxcKafkaData {

    public static void main(String[] args) {


        Properties props = new Properties();
        //用于初始化时建立链接到kafka集群
        props.put("bootstrap.servers", "172.16.2.73:9083,172.16.2.74:9083,172.16.2.75:9083,172.16.2.76:9083,172.16.2.77:9083,172.16.2.78:9083,172.16.2.79:9083");
        //kafka使用消费者分组的概念来允许多个消费者共同消费和处理同一个topic中的消息。分组中消费者成员是动态维护的，如果一个消费者处理失败了，那么之前分配给它的partition将被重新分配给分组中其他消费者；同样，如果分组中加入了新的消费者，也将触发整个partition的重新分配，每个消费者将尽可能的分配到相同数目的partition，以达到新的均衡状态；
        props.put("group.id", "testjxctrace");
        //用于配置是否自动的提交消费进度
        props.put("enable.auto.commit", "true");
        //用于配置自动提交消费进度的时间
        props.put("auto.commit.interval.ms", "1000");
        //会话超时时长，客户端需要周期性的发送“心跳”到broker，这样broker端就可以判断消费者的状态，如果消费在会话周期时长内未发送心跳，那么该消费者将被判定为dead，那么它之前所消费的partition将会被重新的分配给其他存活的消费者
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", StringDeserializer.class);
        props.put("value.deserializer", DecodeingKafka.class);
        props.put("max.poll.records", 100);

        String topicName = args[0];
        String part = args[1];
        String offset = args[2];
        String opt = args[3];
        ConsumerRecords<String, Object> records = null;
        KafkaConsumer<String, Object> consumer = new KafkaConsumer<>(props);
        consumer.assign(Arrays.asList(new TopicPartition(topicName, Integer.valueOf(part))));
        //不改变当前offset，指定从这个topic和partition的开始位置获取。
//        consumer.seekToBeginning(Arrays.asList(new TopicPartition(topicName, 1)));
        consumer.seek(new TopicPartition(topicName, Integer.valueOf(part)), Integer.valueOf(offset));
        boolean isover = false;
        while (true) {
            records = consumer.poll(1000);
            if(records.isEmpty() || isover){
                break;
            }
            for (ConsumerRecord<String, Object> record : records) {
                System.out.println("=====================================");
                System.out.printf("partition = %d, offset = %d, key = %s, value = %s", record.partition(), record.offset(), record.key(), record.value() + "\n");
                TraceBean jxcSynBean = (TraceBean) record.value();
                System.out.println("操作表：" + jxcSynBean.getTraceMainBean().getTableName());
                System.out.println("出入库单号：" + jxcSynBean.getTraceMainBean().getMainCode());
                if(opt.equals("1")){
                    System.out.println(JSON.toJSONString(jxcSynBean));
                    isover = true;
                    break;
                }

            }
            
        }


    }
}
